/*****************************
msgType消息类型说明:
1，注册broker
2，broker下线
3，正常的消息
serialType序列化方式说明:
1, dxserial
2, json
*******************************/

use std::{borrow::BorrowMut, collections::LinkedList, sync::Arc};

use crate::{
    airport::{self, airport::Airport},
    application::Application,
};

use super::{protocol_handler::ProtocolHandler, MsgType, XID};

use async_trait::async_trait;
use common::protocol::protocol_v1::{ProtocolV1Reader, ProtocolV1Writer};
use protobuf::CodedOutputStream;
use semver::Version;
use tracing::info;

const FIX_SIZE: u32 = 8;

//
// pub fn header_len() -> u32 {
//     FIX_LEN_POS as u32
// }

// ==============

pub struct ProtocolV1Handler {}

impl ProtocolV1Handler {
    pub fn new() -> Self {
        ProtocolV1Handler {}
    }
    fn handle_broker_on(&self, channel_id: i64, data: Vec<u8>) {
        //
    }

    async fn handle_normal_message(&self, channel_id: i64, message: Vec<u8>) {
        let protocol_reader = ProtocolV1Reader::new(&message);

        let receiver_service_key = protocol_reader.receiver();

        let airport = Application::get_airport();

        let message_manager = airport.get_message_manager();

        let xid = protocol_reader.xid();

        let message = Arc::new(message);

        let _is_succeed = message_manager
            .add_message(channel_id, xid, message.clone())
            .await;

        let dxc_manager = airport.get_dxc_manager();

        let service_version = Version::parse(&receiver_service_key.dxc_version).unwrap();

        let dxc = dxc_manager
            .get_dxc_by_name_and_version(&receiver_service_key.dxc_name, &service_version)
            .await;

        dxc.message_in_by_name(&receiver_service_key.service_name, message.as_ref())
            .await;
    }
}

#[async_trait]
impl ProtocolHandler for ProtocolV1Handler {
    fn send_msg(&self, data: Vec<u8>) {}

    async fn handle_msg(&self, airport: &Airport, channel_id: i64, message: Vec<u8>) -> bool {
        let message = if channel_id != 0 && channel_id != 1 {
            let channel_manager = airport.get_channel_manager();
            let channel = channel_manager.get_channel(channel_id).await;
            if channel.is_none() {
                info!("channel {} 不存在...", channel_id);
                return false;
            }
            let channel = channel.unwrap();
            let conn_id = channel.conn_id();
            if conn_id != 0 {
                info!("handle_msg 设置 conn_id = {}", conn_id);
                let mut message = message;
                ProtocolV1Writer::write_conn_id_to_message(&mut message, conn_id);
                message
            } else {
                info!("handle_msg 未设置 conn_id");
                message
            }
        } else {
            if channel_id == 0 {
                info!("handle_msg 未设置 conn_id");
            }
            message
        };

        let protocol_reader: ProtocolV1Reader<'_> = ProtocolV1Reader::new(&message);
        let msg_type = protocol_reader.msg_type();
        if msg_type.is_none() {
            return false;
        }
        let msg_type = msg_type.unwrap();
        match msg_type {
            MsgType::DXCOn => {}
            MsgType::DXCOff => {}
            MsgType::AckNormalMsg => {}
            MsgType::RefuseMsg => {}
            MsgType::NormalMsg | MsgType::RspNormalMsg | MsgType::System => {
                self.handle_normal_message(channel_id, message).await
            }
            _ => {
                info!("v1 消息协议，不处理类型：{:?} 的消息", msg_type)
            }
        }
        true
    }

    fn get_xid(&self, data: &Vec<u8>) -> XID {
        let protocol_reader = ProtocolV1Reader::new(&data);
        protocol_reader.xid()
    }

    fn get_version(&self) -> u16 {
        1
    }
}

pub fn make_ack_message(ack_ids: Vec<i64>) -> Vec<u8> {
    let mut message_size = FIX_SIZE as u32;
    for value in &ack_ids {
        message_size += protobuf::rt::int64_size(3, *value) as u32;
    }

    let mut msg_buffer = Vec::with_capacity(message_size as usize);
    unsafe {
        msg_buffer.set_len(message_size as usize);
        let size_byte = message_size.to_le_bytes();
        let size_ptr = msg_buffer.as_mut_ptr();
        std::ptr::copy_nonoverlapping(size_byte.as_ptr(), size_ptr, 4);
        //
        let msg_type = MsgType::AckNormalMsg as i32;
        let type_bytes = msg_type.to_le_bytes();
        let type_ptr = msg_buffer.as_mut_ptr().offset(4);
        std::ptr::copy_nonoverlapping(type_bytes.as_ptr(), type_ptr, 4);
    }

    let ids_buffer = &mut msg_buffer[FIX_SIZE as usize..];
    let mut os = CodedOutputStream::bytes(ids_buffer);
    for v in &ack_ids {
        os.write_int64(1, *v).unwrap();
    }

    os.flush().unwrap();
    drop(os);

    msg_buffer
}

/**
 * 解析ack 信息
 */
pub fn parse_ack_message(message: &Vec<u8>) -> Option<LinkedList<i64>> {
    // 20 只是
    let ids_bytes = &message[FIX_SIZE as usize..];
    let mut is = protobuf::CodedInputStream::from_bytes(ids_bytes);
    let mut ack_ids = LinkedList::new();
    while let Some(_) = is.read_raw_tag_or_eof().unwrap() {
        let id = is.read_int64();
        if id.is_err() {
            return None;
        }
        ack_ids.push_back(id.unwrap());
    }

    if is.check_eof().is_err() {
        None
    } else {
        Some(ack_ids)
    }
}
